package site.edody.dframe.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Map;
import java.util.stream.Collectors;

@Configuration
public class MQConsumerConfig {
    private final Logger logger = LoggerFactory.getLogger(MQConsumerConfig.class);

    @Bean
    public Map<String, DefaultMQPushConsumer> mqConsumers(
            RocketMQProperties properties,
            ConsumerService consumerService) {

        Map<String, RocketMQProperties.ConsumerConfig> consumerConfigs = properties.getConsumers();
        return consumerConfigs.entrySet().stream()
                .collect(Collectors.toMap(
                        Map.Entry::getKey,
                        entry -> createConsumer(
                                entry.getValue(),
                                properties.getNameServer(),
                                consumerService
                        )
                ));
    }

    private DefaultMQPushConsumer createConsumer(
            RocketMQProperties.ConsumerConfig config,
            String nameServer,
            ConsumerService consumerService) {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(config.getGroup());
        consumer.setNamesrvAddr(nameServer);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        try {
            // 订阅主题和标签
            consumer.subscribe(config.getTopic(), String.join(" || ", config.getTags()));

            // 注册消息监听器
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                return consumerService.processMessages(msgs, config.getTopic());
            });

            consumer.start();
            logger.info("Consumer started: " + config.getGroup() + " for topic: " + config.getTopic());
        } catch (Exception e) {
            throw new RuntimeException("Failed to start consumer: " + config.getGroup(), e);
        }
        return consumer;
    }
}